package com.errol.batchprocessor.batchprocessor;

import com.aliyun.openservices.shade.io.netty.util.concurrent.DefaultThreadFactory;
import com.renjia.entity.AbstractImportBatch;
import com.renjia.service.generic.base.util.DateFormatUtil;
import com.renjia.service.generic.log.LogService;
import com.renjia.service.generic.middleware.batchprocessor.advisor.*;
import com.renjia.service.generic.middleware.batchprocessor.bo.ThreadPoolAndMissionStatusBO;
import com.renjia.service.generic.middleware.batchprocessor.bo.ThreadPoolStatusBO;
import com.renjia.service.generic.middleware.batchprocessor.payload.AbstractBatchPayload;
import com.renjia.service.generic.middleware.batchprocessor.payload.AbstractPiecePayload;
import com.renjia.service.generic.middleware.batchprocessor.payload.AbstractSinglePayload;
import com.renjia.service.generic.middleware.batchprocessor.payload.BatchCounter;
import org.springframework.beans.factory.DisposableBean;

import java.util.Date;
import java.util.List;
import java.util.concurrent.*;

/**
 * @author zhang xiao bin
 * @version v1.0
 * @date 2021/02/23
 * @since v
 **/
public abstract class AbstractBatchProcessorImpl<IE extends AbstractImportBatch, AT extends Enum<?>,
        BE, SE,
        BP extends AbstractBatchPayload<IE, AT, BE, SE>, PP extends AbstractPiecePayload<SE, SP>, SP extends AbstractSinglePayload<SE>>
        implements DisposableBean, BatchProcessor<IE, AT, BE, SE, BP, PP, SP> {

    private final ThreadPoolExecutor executor;
    private final ConcurrentMap<String, BatchCounter> batchCounterMap;
    private final BatchAddAdvisor<IE, AT, BE, SE, BP, PP, SP> batchAddAdvisor;
    private final BatchStartAdvisor<IE, AT, BE, SE, BP, PP, SP> batchStartAdvisor;
    private final BatchFinishAdvisor<IE, AT, BE, SE, BP, PP, SP> batchFinishAdvisor;
    private final BatchPostAdvisor<IE, AT, BE, SE, BP, PP, SP> batchPostAdvisor;
    private final PieceStartAdvisor<IE, AT, BE, SE, BP, PP, SP> pieceStartAdvisor;
    private final PieceFinishAdvisor<IE, AT, BE, SE, BP, PP, SP> pieceFinishAdvisor;
    private final PiecePostAdvisor<IE, AT, BE, SE, BP, PP, SP> piecePostAdvisor;
    private final SingleStartAdvisor<IE, AT, BE, SE, BP, PP, SP> singleStartAdvisor;
    private final SingleProcessAdvisor<IE, AT, BE, SE, BP, PP, SP> singleProcessAdvisor;
    private final SingleFinishAdvisor<IE, AT, BE, SE, BP, PP, SP> singleFinishAdvisor;
    private final SinglePostAdvisor<IE, AT, BE, SE, BP, PP, SP> singlePostAdvisor;
    private final LogService logService;

    public AbstractBatchProcessorImpl(BatchAddAdvisor<IE, AT, BE, SE, BP, PP, SP> batchAddAdvisor,
                                      BatchStartAdvisor<IE, AT, BE, SE, BP, PP, SP> batchStartAdvisor,
                                      BatchFinishAdvisor<IE, AT, BE, SE, BP, PP, SP> batchFinishAdvisor,
                                      BatchPostAdvisor<IE, AT, BE, SE, BP, PP, SP> batchPostAdvisor,
                                      PieceStartAdvisor<IE, AT, BE, SE, BP, PP, SP> pieceStartAdvisor,
                                      PieceFinishAdvisor<IE, AT, BE, SE, BP, PP, SP> pieceFinishAdvisor,
                                      PiecePostAdvisor<IE, AT, BE, SE, BP, PP, SP> piecePostAdvisor,
                                      SingleStartAdvisor<IE, AT, BE, SE, BP, PP, SP> singleStartAdvisor,
                                      SingleProcessAdvisor<IE, AT, BE, SE, BP, PP, SP> singleProcessAdvisor,
                                      SingleFinishAdvisor<IE, AT, BE, SE, BP, PP, SP> singleFinishAdvisor,
                                      SinglePostAdvisor<IE, AT, BE, SE, BP, PP, SP> singlePostAdvisor,
                                      LogService logService) {
        this.batchAddAdvisor = batchAddAdvisor;
        this.batchStartAdvisor = batchStartAdvisor;
        this.batchFinishAdvisor = batchFinishAdvisor;
        this.batchPostAdvisor = batchPostAdvisor;
        this.pieceStartAdvisor = pieceStartAdvisor;
        this.pieceFinishAdvisor = pieceFinishAdvisor;
        this.piecePostAdvisor = piecePostAdvisor;
        this.singleStartAdvisor = singleStartAdvisor;
        this.singleProcessAdvisor = singleProcessAdvisor;
        this.singleFinishAdvisor = singleFinishAdvisor;
        this.singlePostAdvisor = singlePostAdvisor;
        this.logService = logService;
        // start threadPoolExecutor
        int corePoolSize = (int) (Runtime.getRuntime().availableProcessors() * 1.25);
        this.executor = new ThreadPoolExecutor(corePoolSize, (int) (corePoolSize * 1.5), 1, TimeUnit.HOURS,
                new LinkedBlockingQueue<>(1024), new DefaultThreadFactory(logPrefix()), new ThreadPoolExecutor.AbortPolicy());
        this.batchCounterMap = new ConcurrentHashMap<>();
        // TODO: 21/2/23 根据内存占用估算队列长度，添加拒绝时的处理逻辑
    }

    @Override
    public void destroy() throws Exception {
        this.executor.shutdown();
        while (this.executor.isTerminating()) {
            Thread.sleep(10);
        }
        // TODO: 21/2/25 等待已开启的 batch 处理完成，未开启的批次等重启后继续
    }

    /**
     * 添加一个批次到线程池，执行以下逻辑：
     *
     * @param batchPayload 批次数据
     */
    @Override
    public void startMission(BP batchPayload) {
        List<PP> piecePayloadList = doSplitBatch(batchPayload);
        batchCounterMap.put(batchPayload.getBatchNo(), new BatchCounter(piecePayloadList.size()));
        piecePayloadList.forEach(piecePayload ->
                this.executor.execute(() -> {
                    boolean isLastPiece;
                    BatchCounter batchCounter = batchCounterMap.get(batchPayload.getBatchNo());
                    try {
                        // 由第一个分片执行批次启动的扩展点；其他分片将会等待，等待4秒后超时
                        doStartBatchByFirstPiece(batchCounter, batchPayload);

                        // 如果批次成功启动，则尝试启动分片；如果分片启动成功，则每条记录都会执行，除非处理异常的扩展中强制抛出异常
                        doProcessPieceIfBatchStarted(batchCounter, piecePayload, batchPayload);

                        // 分片正常结束（执行成功、或未执行）
                        isLastPiece = batchCounter.finishOneMissionPiece();
                    } catch (Exception e) {
                        logService.error(e, logPrefix() + "启动批次或执行分片时异常");
                        // 分片抛出异常，重置分片状态
                        isLastPiece = batchCounter.interruptOneMissionPiece();
                        batchCounter.setLastPieceErrorMessageHint(getErrorMessageHint(e));
                        try {
                            pieceFinishAdvisor.doRecoverOnPieceException(piecePayload, batchPayload, batchCounter.getLastPieceErrorMessageHint());
                        } catch (Exception ex) {
                            logService.error(ex, logPrefix() + "重置分片时异常");
                        }
                    } finally {
                        // 如果批次成功启动，才会启动分片，即便是启动批次的分片
                        if (batchCounter.isBatchStartSuccess()) {
                            try {
                                // 分片执行完成、启动失败、或执行异常，都会清理分片缓存和资源
                                piecePostAdvisor.doCleanUpOnPiecePost(piecePayload, batchPayload);
                            } catch (Exception e) {
                                logService.error(e, logPrefix() + "清理分片缓存和资源时异常");
                            }
                        }
                    }

                    if (isLastPiece) {
                        // 如果是最后一个分片，则执行结束批次的扩展
                        batchCounterMap.remove(batchPayload.getBatchNo());
                        doFinishBatchByLastPiece(batchCounter, batchPayload);
                    }
                })
        );
    }

    @Override
    public List<PP> doSplitBatch(BP batchPayload) {
        return batchAddAdvisor.doSplitOnBatchAdded(batchPayload, getThreadPoolStatus());
    }

    /**
     * 启动一个分片
     * - 如果当前分片是批次的第一个分片，执行批次的启动扩展；该扩展可以主动拦截批次，此时所有分片都不会执行；启动异常则整个批次异常
     * - 如果不是第一个分片，则等待批次启动完成；启动成功则分片执行，否则不执行；最多等待4秒，超时则分片异常
     *
     * @param batchCounter 批次计数器
     * @param batchPayload 批次数据
     */
    private void doStartBatchByFirstPiece(BatchCounter batchCounter, BP batchPayload) throws InterruptedException {
        if (batchCounter.startOneMissionPiece()) {
            // 批次中第一个分片，尝试启动批次、准备数据；返回 true 表示批次启动成功、false 表示主动拦截；抛出异常则批次失败，需要重试
            try {
                if (batchStartAdvisor.doPrepareOnBatchStart(batchPayload)) {
                    batchCounter.startBatchSuccess();
                } else {
                    batchCounter.startBatchFailed();
                }
            } catch (Exception e) {
                batchCounter.startBatchFailed();
                logService.remind(logPrefix() + "任务批次启动异常（注意：该扩展点不应抛出异常），batchNo=" + batchPayload.getBatchNo());
                throw e;
            }
        } else if (batchCounter.isBatchStarting()) {
            // 批次中非第一个分片，则需等待批次启动完成（成功或失败）；最多等待 4 秒，超时抛出异常，分片失败，批次需要重试
            int c = 128;
            do {
                if (c-- < 0) {
                    logService.remind(logPrefix() + "任务分片等待批次启动超时（4秒），batchNo=" + batchPayload.getBatchNo());
                    throw new RuntimeException();
                }
                Thread.sleep(16);
            } while (batchCounter.isBatchStarting());
            logService.debug(logPrefix() + "piece 等待了 " + (16 * (128 - c)) + " ms");
        }
    }

    private void doProcessPieceIfBatchStarted(BatchCounter batchCounter, PP piecePayload, BP batchPayload) {
        if (batchCounter.isBatchStartSuccess()) {
            // 批次启动成功，则尝试启动当前分片、准备数据；返回 true 表示启动成功、false 表示主动拦截；抛出异常则分片失败，批次需要重试
            if (pieceStartAdvisor.doPrepareOnPieceStart(piecePayload, batchPayload)) {
                for (SP singlePayload : piecePayload.getSinglePayloadList()) {
                    // 分片启动成功，循环处理每条记录；某条记录处理异常不会导致分片异常，如果希望分片异常可以在 catch 回调中主动抛出异常
                    doProcessSingleRecord(singlePayload, piecePayload, batchPayload);
                }
                // 分片启动成功、且中途未抛出异常，对分片进行结算
                pieceFinishAdvisor.doSettleOnPieceProcessed(piecePayload, batchPayload);
            }
        }
    }

    @Override
    public boolean doProcessSingleRecord(SP singlePayload, PP piecePayload, BP batchPayload) {
        try {
            if (singleStartAdvisor.doPrepareOnSingleStart(singlePayload, piecePayload, batchPayload)) {
                // 一条记录的数据准备成功，才进行校验；校验成功，才进行处理；所有处理扩展点都会执行，执行未抛出异常，才进行结算，否则进行重置
                if (singleProcessAdvisor.doValidateBeforeSingleProcess(singlePayload, piecePayload, batchPayload)) {
                    singleProcessAdvisor.doExecuteOnSingleProcess(singlePayload, piecePayload, batchPayload);
                }
                singleFinishAdvisor.doSettleOnSingleProcessed(singlePayload, piecePayload, batchPayload);
            }
            return true;
        } catch (Exception e) {
            logService.error(e, logPrefix() + "处理单条记录时异常");
            singleFinishAdvisor.doRecoverOnSingleException(singlePayload, piecePayload, batchPayload, getErrorMessageHint(e));
            return false;
        } finally {
            try {
                singlePostAdvisor.doCleanUpOnSinglePost(singlePayload, piecePayload, batchPayload);
            } catch (Exception e) {
                logService.error(e, logPrefix() + "处理完一条数据后清理缓存异常");
            }
        }
    }

    private void doFinishBatchByLastPiece(BatchCounter batchCounter, BP batchPayload) {
        try {
            // 批次启动、且所有分片执行成功，对整个批次进行结算
            if (batchCounter.isBatchStartSuccess()
                    && batchCounter.isAllPiecesSuccess()) {
                batchFinishAdvisor.doSettleOnBatchProcessed(batchPayload);
            }
            // 批次启动异常、或存在分片执行异常，则批次执行失败，需择机重试
            if (!batchCounter.isAllPiecesSuccess()) {
                batchFinishAdvisor.doRecoverOnBatchException(batchPayload, batchCounter.getLastPieceErrorMessageHint());
                // TODO: 21/2/24 自动重试的逻辑
            }
            // 除以上情况外，如果批次未启动、且无分片异常，则表示批次被主动拦截，不做批次结束的处理
        } catch (Exception e) {
            logService.error(e, logPrefix() + "结算或重置批次时异常");
        } finally {
            doCleanBatch(batchPayload);
        }
    }

    @Override
    public void doCleanBatch(BP batchPayload) {
        try {
            // 批次执行完成，清理批次缓存和资源
            batchPostAdvisor.doCleanUpOnBatchPost(batchPayload);
        } catch (Exception e) {
            logService.error(e, logPrefix() + "清理批次缓存和资源时异常");
        }
    }

    private String getErrorMessageHint(Exception e) {
        return DateFormatUtil.date2String(new Date(), DateFormatUtil.PATTERN_A) + " " + e.getMessage();
    }

    /**
     * 输出日志时的前缀
     * @return 输出日志时的前缀
     */
    protected abstract String logPrefix();

    @Override
    public ThreadPoolStatusBO getThreadPoolStatus() {
        return new ThreadPoolStatusBO(executor);
    }

    @Override
    public ThreadPoolAndMissionStatusBO getMissionStatus() {
        return new ThreadPoolAndMissionStatusBO(executor, batchCounterMap);
    }

    @Override
    public int setCorePoolSize(int targetCorePoolSize) {
        targetCorePoolSize = Math.min(targetCorePoolSize, Runtime.getRuntime().availableProcessors() * 5);
        if (targetCorePoolSize > executor.getCorePoolSize()) {
            executor.setCorePoolSize((int) Math.min(targetCorePoolSize, executor.getCorePoolSize() * 1.5));
        } else {
            executor.setCorePoolSize((int) Math.max(targetCorePoolSize, executor.getCorePoolSize() * 0.5));
        }
        executor.setMaximumPoolSize((int) (executor.getCorePoolSize() * 1.5));
        return executor.getCorePoolSize();
    }
}
